{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "source": [
        "import google.generativeai as genai\n",
        "import json\n",
        "import time\n",
        "from typing import Dict, List, Any, Optional\n",
        "from dataclasses import dataclass, asdict\n",
        "from enum import Enum\n",
        "\n",
        "class TaskStatus(Enum):\n",
        "    PENDING = \"pending\"\n",
        "    IN_PROGRESS = \"in_progress\"\n",
        "    COMPLETED = \"completed\"\n",
        "    FAILED = \"failed\""
      ],
      "metadata": {
        "id": "FEFZuu2vEHi5"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "@dataclass\n",
        "class Task:\n",
        "    id: str\n",
        "    description: str\n",
        "    priority: int\n",
        "    status: TaskStatus = TaskStatus.PENDING\n",
        "    dependencies: List[str] = None\n",
        "    result: Optional[str] = None\n",
        "\n",
        "    def __post_init__(self):\n",
        "        if self.dependencies is None:\n",
        "            self.dependencies = []\n",
        "\n",
        "class SAGEAgent:\n",
        "    \"\"\"Self-Adaptive Goal-oriented Execution AI Agent\"\"\"\n",
        "\n",
        "    def __init__(self, api_key: str, model_name: str = \"gemini-1.5-flash\"):\n",
        "        genai.configure(api_key=api_key)\n",
        "        self.model = genai.GenerativeModel(model_name)\n",
        "        self.memory = []\n",
        "        self.tasks = {}\n",
        "        self.context = {}\n",
        "        self.iteration_count = 0\n",
        "\n",
        "    def self_assess(self, goal: str, context: Dict[str, Any]) -> Dict[str, Any]:\n",
        "        \"\"\"S: Self-Assessment - Evaluate current state and capabilities\"\"\"\n",
        "        assessment_prompt = f\"\"\"\n",
        "        You are an AI agent conducting self-assessment. Respond ONLY with valid JSON, no additional text.\n",
        "\n",
        "        GOAL: {goal}\n",
        "        CONTEXT: {json.dumps(context, indent=2)}\n",
        "        TASKS_PROCESSED: {len(self.tasks)}\n",
        "\n",
        "        Provide assessment as JSON with these exact keys:\n",
        "        {{\n",
        "            \"progress_score\": <number 0-100>,\n",
        "            \"resources\": [\"list of available resources\"],\n",
        "            \"gaps\": [\"list of knowledge gaps\"],\n",
        "            \"risks\": [\"list of potential risks\"],\n",
        "            \"recommendations\": [\"list of next steps\"]\n",
        "        }}\n",
        "        \"\"\"\n",
        "\n",
        "        response = self.model.generate_content(assessment_prompt)\n",
        "        try:\n",
        "            text = response.text.strip()\n",
        "            if text.startswith('```'):\n",
        "                text = text.split('```')[1]\n",
        "                if text.startswith('json'):\n",
        "                    text = text[4:]\n",
        "            text = text.strip()\n",
        "            return json.loads(text)\n",
        "        except Exception as e:\n",
        "            print(f\"Assessment parsing error: {e}\")\n",
        "            return {\n",
        "                \"progress_score\": 25,\n",
        "                \"resources\": [\"AI capabilities\", \"Internet knowledge\"],\n",
        "                \"gaps\": [\"Specific domain expertise\", \"Real-time data\"],\n",
        "                \"risks\": [\"Information accuracy\", \"Scope complexity\"],\n",
        "                \"recommendations\": [\"Break down into smaller tasks\", \"Focus on research first\"]\n",
        "            }\n",
        "\n",
        "    def adaptive_plan(self, goal: str, assessment: Dict[str, Any]) -> List[Task]:\n",
        "        \"\"\"A: Adaptive Planning - Create dynamic, context-aware task decomposition\"\"\"\n",
        "        planning_prompt = f\"\"\"\n",
        "        You are an AI task planner. Respond ONLY with valid JSON array, no additional text.\n",
        "\n",
        "        MAIN_GOAL: {goal}\n",
        "        ASSESSMENT: {json.dumps(assessment, indent=2)}\n",
        "\n",
        "        Create 3-4 actionable tasks as JSON array:\n",
        "        [\n",
        "            {{\n",
        "                \"id\": \"task_1\",\n",
        "                \"description\": \"Clear, specific task description\",\n",
        "                \"priority\": 5,\n",
        "                \"dependencies\": []\n",
        "            }},\n",
        "            {{\n",
        "                \"id\": \"task_2\",\n",
        "                \"description\": \"Another specific task\",\n",
        "                \"priority\": 4,\n",
        "                \"dependencies\": [\"task_1\"]\n",
        "            }}\n",
        "        ]\n",
        "\n",
        "        Each task must have: id (string), description (string), priority (1-5), dependencies (array of strings)\n",
        "        \"\"\"\n",
        "\n",
        "        response = self.model.generate_content(planning_prompt)\n",
        "        try:\n",
        "            text = response.text.strip()\n",
        "            if text.startswith('```'):\n",
        "                text = text.split('```')[1]\n",
        "                if text.startswith('json'):\n",
        "                    text = text[4:]\n",
        "            text = text.strip()\n",
        "\n",
        "            task_data = json.loads(text)\n",
        "            tasks = []\n",
        "            for i, task_info in enumerate(task_data):\n",
        "                task = Task(\n",
        "                    id=task_info.get('id', f'task_{i+1}'),\n",
        "                    description=task_info.get('description', 'Undefined task'),\n",
        "                    priority=task_info.get('priority', 3),\n",
        "                    dependencies=task_info.get('dependencies', [])\n",
        "                )\n",
        "                tasks.append(task)\n",
        "            return tasks\n",
        "        except Exception as e:\n",
        "            print(f\"Planning parsing error: {e}\")\n",
        "            return [\n",
        "                Task(id=\"research_1\", description=\"Research sustainable urban gardening basics\", priority=5),\n",
        "                Task(id=\"research_2\", description=\"Identify space-efficient growing methods\", priority=4),\n",
        "                Task(id=\"compile_1\", description=\"Organize findings into structured guide\", priority=3, dependencies=[\"research_1\", \"research_2\"])\n",
        "            ]\n",
        "\n",
        "    def execute_goal_oriented(self, task: Task) -> str:\n",
        "        \"\"\"G: Goal-oriented Execution - Execute specific task with focused attention\"\"\"\n",
        "        execution_prompt = f\"\"\"\n",
        "        GOAL-ORIENTED EXECUTION:\n",
        "        Task: {task.description}\n",
        "        Priority: {task.priority}\n",
        "        Context: {json.dumps(self.context, indent=2)}\n",
        "\n",
        "        Execute this task step-by-step:\n",
        "        1. Break down the task into concrete actions\n",
        "        2. Execute each action methodically\n",
        "        3. Validate results at each step\n",
        "        4. Provide comprehensive output\n",
        "\n",
        "        Focus on practical, actionable results. Be specific and thorough.\n",
        "        \"\"\"\n",
        "\n",
        "        response = self.model.generate_content(execution_prompt)\n",
        "        return response.text.strip()\n",
        "\n",
        "    def integrate_experience(self, task: Task, result: str, success: bool) -> Dict[str, Any]:\n",
        "        \"\"\"E: Experience Integration - Learn from outcomes and update knowledge\"\"\"\n",
        "        integration_prompt = f\"\"\"\n",
        "        You are learning from task execution. Respond ONLY with valid JSON, no additional text.\n",
        "\n",
        "        TASK: {task.description}\n",
        "        RESULT: {result[:200]}...\n",
        "        SUCCESS: {success}\n",
        "\n",
        "        Provide learning insights as JSON:\n",
        "        {{\n",
        "            \"learnings\": [\"key insight 1\", \"key insight 2\"],\n",
        "            \"patterns\": [\"pattern observed 1\", \"pattern observed 2\"],\n",
        "            \"adjustments\": [\"adjustment for future 1\", \"adjustment for future 2\"],\n",
        "            \"confidence_boost\": <number -10 to 10>\n",
        "        }}\n",
        "        \"\"\"\n",
        "\n",
        "        response = self.model.generate_content(integration_prompt)\n",
        "        try:\n",
        "            text = response.text.strip()\n",
        "            if text.startswith('```'):\n",
        "                text = text.split('```')[1]\n",
        "                if text.startswith('json'):\n",
        "                    text = text[4:]\n",
        "            text = text.strip()\n",
        "\n",
        "            experience = json.loads(text)\n",
        "            experience['task_id'] = task.id\n",
        "            experience['timestamp'] = time.time()\n",
        "            self.memory.append(experience)\n",
        "            return experience\n",
        "        except Exception as e:\n",
        "            print(f\"Experience parsing error: {e}\")\n",
        "            experience = {\n",
        "                \"learnings\": [f\"Completed task: {task.description}\"],\n",
        "                \"patterns\": [\"Task execution follows planned approach\"],\n",
        "                \"adjustments\": [\"Continue systematic approach\"],\n",
        "                \"confidence_boost\": 5 if success else -2,\n",
        "                \"task_id\": task.id,\n",
        "                \"timestamp\": time.time()\n",
        "            }\n",
        "            self.memory.append(experience)\n",
        "            return experience\n",
        "\n",
        "    def execute_sage_cycle(self, goal: str, max_iterations: int = 3) -> Dict[str, Any]:\n",
        "        \"\"\"Execute complete SAGE cycle for goal achievement\"\"\"\n",
        "        print(f\"🎯 Starting SAGE cycle for goal: {goal}\")\n",
        "        results = {\"goal\": goal, \"iterations\": [], \"final_status\": \"unknown\"}\n",
        "\n",
        "        for iteration in range(max_iterations):\n",
        "            self.iteration_count += 1\n",
        "            print(f\"\\n🔄 SAGE Iteration {iteration + 1}\")\n",
        "\n",
        "            print(\"📊 Self-Assessment...\")\n",
        "            assessment = self.self_assess(goal, self.context)\n",
        "            print(f\"Progress Score: {assessment.get('progress_score', 0)}/100\")\n",
        "\n",
        "            print(\"🗺️  Adaptive Planning...\")\n",
        "            tasks = self.adaptive_plan(goal, assessment)\n",
        "            print(f\"Generated {len(tasks)} tasks\")\n",
        "\n",
        "            print(\"⚡ Goal-oriented Execution...\")\n",
        "            iteration_results = []\n",
        "\n",
        "            for task in sorted(tasks, key=lambda x: x.priority, reverse=True):\n",
        "                if self._dependencies_met(task):\n",
        "                    print(f\"  Executing: {task.description}\")\n",
        "                    task.status = TaskStatus.IN_PROGRESS\n",
        "\n",
        "                    try:\n",
        "                        result = self.execute_goal_oriented(task)\n",
        "                        task.result = result\n",
        "                        task.status = TaskStatus.COMPLETED\n",
        "                        success = True\n",
        "                        print(f\"  ✅ Completed: {task.id}\")\n",
        "                    except Exception as e:\n",
        "                        task.status = TaskStatus.FAILED\n",
        "                        task.result = f\"Error: {str(e)}\"\n",
        "                        success = False\n",
        "                        print(f\"  ❌ Failed: {task.id}\")\n",
        "\n",
        "                    experience = self.integrate_experience(task, task.result, success)\n",
        "\n",
        "                    self.tasks[task.id] = task\n",
        "                    iteration_results.append({\n",
        "                        \"task\": asdict(task),\n",
        "                        \"experience\": experience\n",
        "                    })\n",
        "\n",
        "            self._update_context(iteration_results)\n",
        "\n",
        "            results[\"iterations\"].append({\n",
        "                \"iteration\": iteration + 1,\n",
        "                \"assessment\": assessment,\n",
        "                \"tasks_generated\": len(tasks),\n",
        "                \"tasks_completed\": len([r for r in iteration_results if r[\"task\"][\"status\"] == \"completed\"]),\n",
        "                \"results\": iteration_results\n",
        "            })\n",
        "\n",
        "            if assessment.get('progress_score', 0) >= 90:\n",
        "                results[\"final_status\"] = \"achieved\"\n",
        "                print(\"🎉 Goal achieved!\")\n",
        "                break\n",
        "\n",
        "        if results[\"final_status\"] == \"unknown\":\n",
        "            results[\"final_status\"] = \"in_progress\"\n",
        "\n",
        "        return results\n",
        "\n",
        "    def _dependencies_met(self, task: Task) -> bool:\n",
        "        \"\"\"Check if task dependencies are satisfied\"\"\"\n",
        "        for dep_id in task.dependencies:\n",
        "            if dep_id not in self.tasks or self.tasks[dep_id].status != TaskStatus.COMPLETED:\n",
        "                return False\n",
        "        return True\n",
        "\n",
        "    def _update_context(self, results: List[Dict[str, Any]]):\n",
        "        \"\"\"Update agent context based on execution results\"\"\"\n",
        "        completed_tasks = [r for r in results if r[\"task\"][\"status\"] == \"completed\"]\n",
        "        self.context.update({\n",
        "            \"completed_tasks\": len(completed_tasks),\n",
        "            \"total_tasks\": len(self.tasks),\n",
        "            \"success_rate\": len(completed_tasks) / len(results) if results else 0,\n",
        "            \"last_update\": time.time()\n",
        "        })"
      ],
      "metadata": {
        "id": "8H6w75pnELro"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 974
        },
        "id": "6I10Vv00Jhyy",
        "outputId": "a74d5348-602c-4b7e-a151-d0bc4167ba3f"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "🎯 Starting SAGE cycle for goal: Research and create a comprehensive guide on sustainable urban gardening practices\n",
            "\n",
            "🔄 SAGE Iteration 1\n",
            "📊 Self-Assessment...\n",
            "Progress Score: 0/100\n",
            "🗺️  Adaptive Planning...\n",
            "Generated 4 tasks\n",
            "⚡ Goal-oriented Execution...\n",
            "  Executing: Conduct a literature review on sustainable urban gardening practices, focusing on composting, water conservation, and pest control techniques.\n",
            "  ✅ Completed: task_1\n",
            "  Executing: Research legal and regulatory frameworks governing urban gardening in various locations (e.g., zoning laws, water rights).\n",
            "  ✅ Completed: task_2\n",
            "  Executing: Identify and interview experts in sustainable urban gardening and relevant fields (e.g., urban planning, environmental science, community development).\n",
            "  ✅ Completed: task_3\n",
            "  Executing: Compile research findings and draft initial sections of the guide, including case studies and practical examples.\n",
            "  ✅ Completed: task_4\n",
            "\n",
            "🔄 SAGE Iteration 2\n",
            "📊 Self-Assessment...\n",
            "Progress Score: 25/100\n",
            "🗺️  Adaptive Planning...\n",
            "Generated 4 tasks\n",
            "⚡ Goal-oriented Execution...\n",
            "  Executing: Conduct comprehensive research on sustainable urban gardening techniques, including composting, water harvesting, and pest control methods. Compile a list of reliable sources and relevant data.\n",
            "  ✅ Completed: task_1\n",
            "  Executing: Develop a detailed outline for the guide, incorporating researched techniques, case studies, legal frameworks, and sections on soil health, biodiversity, and community engagement.\n",
            "  ✅ Completed: task_4\n",
            "  Executing: Identify and analyze at least five case studies of successful urban gardening initiatives in diverse urban environments. Document best practices and challenges faced.\n",
            "  ✅ Completed: task_2\n",
            "  Executing: Research and summarize legal and regulatory frameworks governing urban gardening in three different countries or regions.  Identify potential obstacles and opportunities.\n",
            "  ✅ Completed: task_3\n",
            "\n",
            "==================================================\n",
            "📋 SAGE EXECUTION SUMMARY\n",
            "==================================================\n",
            "Goal: Research and create a comprehensive guide on sustainable urban gardening practices\n",
            "Status: in_progress\n",
            "Iterations: 2\n",
            "\n",
            "Iteration 1:\n",
            "  Assessment Score: 0/100\n",
            "  Tasks Generated: 4\n",
            "  Tasks Completed: 0\n",
            "\n",
            "Iteration 2:\n",
            "  Assessment Score: 25/100\n",
            "  Tasks Generated: 4\n",
            "  Tasks Completed: 0\n",
            "\n",
            "🧠 Agent Memory Entries: 8\n",
            "🎯 Total Tasks Processed: 4\n"
          ]
        }
      ],
      "source": [
        "if __name__ == \"__main__\":\n",
        "    API_KEY = \"Use Your Own API Key Here\"\n",
        "\n",
        "    try:\n",
        "        agent = SAGEAgent(API_KEY, model_name=\"gemini-1.5-flash\")\n",
        "\n",
        "        goal = \"Research and create a comprehensive guide on sustainable urban gardening practices\"\n",
        "\n",
        "        results = agent.execute_sage_cycle(goal, max_iterations=2)\n",
        "\n",
        "        print(\"\\n\" + \"=\"*50)\n",
        "        print(\"📋 SAGE EXECUTION SUMMARY\")\n",
        "        print(\"=\"*50)\n",
        "        print(f\"Goal: {results['goal']}\")\n",
        "        print(f\"Status: {results['final_status']}\")\n",
        "        print(f\"Iterations: {len(results['iterations'])}\")\n",
        "\n",
        "        for i, iteration in enumerate(results['iterations'], 1):\n",
        "            print(f\"\\nIteration {i}:\")\n",
        "            print(f\"  Assessment Score: {iteration['assessment'].get('progress_score', 0)}/100\")\n",
        "            print(f\"  Tasks Generated: {iteration['tasks_generated']}\")\n",
        "            print(f\"  Tasks Completed: {iteration['tasks_completed']}\")\n",
        "\n",
        "        print(\"\\n🧠 Agent Memory Entries:\", len(agent.memory))\n",
        "        print(\"🎯 Total Tasks Processed:\", len(agent.tasks))\n",
        "\n",
        "    except Exception as e:\n",
        "        print(f\"Demo requires valid Gemini API key. Error: {e}\")\n",
        "        print(\"Get your free API key from: https://makersuite.google.com/app/apikey\")"
      ]
    }
  ]
}